/******************************************************************************************************************
terminology:
  1. incubation: after a replica file is created, it needs an
  incubation procedure to fill data, before it can be used
  2. journal-based incubation: if a replica is down for some reason,
  and it goes up again in a short time, it can be
     increamenally incubated, instead of fully
  3. write suspended chunks: temporarily unwritable, between a
  replica of the chunk is incubated, and it is available
  4. new replica list: a newly available replica, it is unknown to
  master and client, so the primary needs to record
     it and notfity client when next time the chunk is written
******************************************************************************************************************/

#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <limits.h>
#include <sys/stat.h>
#include <string.h>
#include <st.h>
#include <time.h>
#define USE_LOCAL_LOG_LEVEL
#include "ursax.h"
#include "log.h"
#include "utcache.h"
#include "incubator.h"
#include "redis-client.h"
#include "heartbeat.h"
#include "range-lock.h"
#include "protocol.h"
#include "backend-fs.h"
#include "protocol.h"
#include "concurrent-connection-pool.h"

#define DEBUG_INCUBATE DEBUG_FUNC(st_usleep(100*1000))

static struct foo_cache* write_suspended_chunks = NULL;
static struct foo_hashtable* new_replicas = NULL;

static inline int _sprintf_chunk_filename(char* buf, uint32_t volumeid, uint32_t index,
		const char* prefix)
{
	return sprintf(buf, "%s/%08x.%u", prefix, volumeid, index);
}

static inline int sprintf_incubating_path_filename(char* buf, uint32_t volumeid,
		uint32_t index)
{
	return _sprintf_chunk_filename(buf, volumeid, index, "being_incubated");
}

void destroy_cond(void* _cond)
{
	st_cond_t cond = (st_cond_t)_cond;
	while(st_cond_destroy(cond) != 0) {
		if (unlikely(errno != EBUSY)) {
			LOG_ERROR("condition variable destroy failed!");
			return;
		}
		// yield to let other threads to finish using this cond, and try again
		LOG_WARN("st_cond_destroy ERROR BUSY");
		st_usleep(0);
	}
	LOG_DEBUG("%s function end", __func__);
}

static inline int sprintf_incubating_relative_path_filename(char* buf, uint32_t volumeid,
        uint32_t index)
{
	int nbytes = sprintf(buf, "../");
	nbytes += sprintf_chunk_path(buf + nbytes, volumeid, index);
	return sprintf(buf + nbytes, "%08x.%u", volumeid, index);
}

struct NewReplicaList
{
	uint32_t ip;
	uint16_t port;
	time_t breed_time;
	struct NewReplicaList* next;
};

static void destroy_replica_list(void* node)
{
	struct NewReplicaList* r = (struct NewReplicaList*)node;
	while(r) {
		struct NewReplicaList* next = r->next;
		free (r);
		r = next;
	}
}

int incubator_init()
{
	LOG_DEBUG("incubator_init");
	write_suspended_chunks = foo_cache_create(10000000, &destroy_cond, false);
	new_replicas = foo_hashtable_create(&destroy_replica_list);
	return 0;
}

void incubator_fini()
{
	foo_cache_delete(write_suspended_chunks, 0);
	foo_hashtable_delete(new_replicas, 0);
}

bool be_chunk_update_being_incubated(struct Chunk* chunk)
{
	LOG_DEBUG("%s(chunk=%08x.%u)", __func__, chunk->volumeid, chunk->index);

	char path[PATH_MAX], buf[PATH_MAX];
	sprintf_incubating_path_filename(path, chunk->volumeid, chunk->index);
	int ret = readlink(path, buf, sizeof(buf));
	if (likely(ret < 0))
	{
		LOG_DEBUG("reading symlink '%s' for incubation failed %d: %s", path, errno, strerror(errno));
		return chunk->being_incubated = false;
	}
	buf[ret] = '\0';
	sprintf_incubating_relative_path_filename(path, chunk->volumeid, chunk->index);
	if (unlikely(strcmp(buf, path) != 0))
	{
		LOG_ERROR("incubation symlink exists for chunk (%u, %u), but it points to something wrong '%s'",
				  chunk->volumeid, chunk->index, buf);
		return chunk->being_incubated = false;
	}

	LOG_INFO("reading symlink '%s' '%s' exist chunk %08x.%d", path, buf, chunk->volumeid, chunk->index);
	chunk->version = -1;
	return chunk->being_incubated = true;
}

int be_chunk_set_being_incubated(struct Chunk* chunk)
{
	LOG_INFO("%s(chunk=%08x.%u)", __func__, chunk->volumeid, chunk->index);

	char path[PATH_MAX], incubating_path[PATH_MAX];
	sprintf_incubating_relative_path_filename(path, chunk->volumeid, chunk->index);
	sprintf_incubating_path_filename(incubating_path, chunk->volumeid, chunk->index);

	LOG_DEBUG("incubating_relative_path_filename: %s", path);
	LOG_DEBUG("incubating_path_filename: %s", incubating_path);

	int ret = symlink(path, incubating_path);
	if (ret < 0)
	{
		LOG_ERROR("creating symlink ‘%s’ for incubation failed %d: %s", incubating_path, errno, strerror(errno));
	}

	chunk->version = -1;
	ret = be_chunk_write_version(chunk);
	if(ret < 0){
		LOG_ERROR("be_chunk_set_being_incubated be_chunk_write_version ERROR");
	}

	chunk->being_incubated = true;
	heartbeat_remove_chunk(chunk->key);

	chunk->journal.version0 = -1;
	return ret;
}

int add_write_suspended_chunks(uint32_t volumeid, uint32_t index)
{
	st_cond_t cond;
	uint64_t key = chunk_make_key(volumeid, index);
	int ret = foo_cache_get_reference(write_suspended_chunks, key, &cond);
	if(ret < 0) { // not found
		cond = st_cond_new();
		foo_cache_put_reference(write_suspended_chunks, key, cond, 1);
	}
	else {// found
		//reference added in function foo_cache_get_reference, do nothing here
	}
	return 0;
}

void remove_write_suspended_chunks(uint32_t volumeid, uint32_t index)
{
    st_cond_t cond = NULL;
    uint64_t key = chunk_make_key(volumeid, index);
    int ret = foo_cache_adjust_reference(write_suspended_chunks, key, -1);
    if (ret == 0 || (ret < 0 && ret != -EINVAL)) {
        foo_cache_get(write_suspended_chunks, key, &cond);
        st_cond_broadcast(cond);
        st_sleep(0);
        destroy_cond(cond);
    }
}

void remove_new_replica_record(uint32_t volumeid, uint32_t index)
{
    LOG_DEBUG("%s", __func__);
    struct NewReplicaList* p;
    uint64_t key = chunk_make_key(volumeid, index);
    int ret = foo_hashtable_remove(new_replicas, key, &p);
    if (likely(ret < 0))
    {
        // LOG_DEBUG("check_new_replica_record: no new replica record for this chunk");
        return;
    }

    for (struct NewReplicaList* next; p; p = next){
        next = p->next;
        free(p);
    }
    LOG_DEBUG("leave %s", __func__);
}

void _write_check_incubation(struct Chunk* chunk, uint64_t offset, uint64_t size,
        void* buf)
{
    LOG_DEBUG("%s(chunk=%08x.%u, offset=%u, length=%u)",
            __func__, chunk->volumeid, chunk->index, offset, size);

    struct Incubation* incubation = chunk->incubation;
    uint64_t pos = incubation->progress;
    if (offset < pos)
    {
        if (offset + size > pos)
            size = pos - offset;

        int ret = incubation->chunkwrite(
                // todo check (pooled_chunk_write -> chunk_write)
                incubation->target, chunk->volumeid, chunk->index, -1,
                // todo check 0 -> -1
                offset, buf, size);
        if (ret < 0)
        {
            LOG_ERROR("write_check_incubation: failed to write data to "
                      "remote replica (ip %s, port %u, volume %u, index %u, "
                      "offset %u, length %u)",
                      str_ip(incubation->to.ip), incubation->to.port,
                      chunk->volumeid, chunk->index, offset, size);
            // todo: what to do ?
        }
    }
}

int check_wait_write_suspended(uint32_t volumeid, uint32_t index)
{
    LOG_DEBUG("%s(chunk=%08x.%u)", __func__, volumeid, index);

    st_cond_t cond;
    uint64_t key = chunk_make_key(volumeid, index);
    int ret = foo_cache_get_reference(write_suspended_chunks, key, &cond);
    if (unlikely(ret > 0))		// the chunk is suspended
    {
        LOG_INFO("check_wait_write_suspended chunk %08x.%d ref is %d",
                volumeid, index, ret-1);
        foo_cache_adjust_reference(write_suspended_chunks, key, -1);
        again:
        ret = st_cond_timedwait(cond, URSAX_TIMEOUT/10*8);
        if (ret < 0)
        {
            LOG_ERROR("condition wait failed with %d: %s", errno, strerror(errno));
            if (errno == EINTR) goto again;
            if (errno == ETIME){
                LOG_ERROR("error chunk %08x.%d wait %ds timeout, "
                          "unsuspend must be something wrong!!!, dec refrence here",
                          volumeid, index, URSAX_TIMEOUT/2);
                int ref = foo_cache_adjust_reference(write_suspended_chunks, key, 0);
                if(ref == -EINVAL){
                    LOG_ERROR("error foo_cache_adjust_reference is -EINVAL "
                              "volumeid %x index %d, just ret", volumeid, index);
                }else{
                    LOG_ERROR("error foo_cache_adjust_reference 0 ret %d !!!!, "
                              "remove_write_suspended_chunks once here chunk %08x.%d",
                              ref, volumeid, index);
                    remove_write_suspended_chunks(volumeid, index);
                }
            }
            LOG_WARN("check_wait_write_suspended chunk %08x.%d wait over timeout",
                    volumeid, index);
            return -1;
        }
        LOG_INFO("check_wait_write_suspended chunk %08x.%d wait over normally",
                volumeid, index);
        return 1;
    }
    return 0;
}

#define NEW_RECORD_TIMEOUT 60
bool check_new_replica_record(struct CSEP servers[], uint32_t volumeid, uint32_t index)
{
    // LOG_DEBUG(__func__);
    struct NewReplicaList* first;
    uint64_t key = chunk_make_key(volumeid, index);
    int ret = foo_hashtable_get(new_replicas, key, &first);
    if (likely(ret < 0))
    {
        // LOG_DEBUG("check_new_replica_record: no new replica record for this chunk");
        return false;
    }

    int n;	// number of replicas in the request
    for (n=0; n<MAX_FOLLOWING_SERVERS; ++n)
    {
        LOG_DEBUG("ip %s port %d", str_ip(servers[n].ip), servers[n].port);
        if (servers[n].ip == 0)
            break;
    }
    LOG_DEBUG("%s client know %d servers", __func__, n);

    // mark all items that the client has konwn
    struct NewReplicaList* p;
    for (p=first; p; p=p->next)
    {
        LOG_DEBUG("%s new list one %s %d", __func__, str_ip(p->ip), p->port);
        for (int i=0; i<n; ++i)
        {
            if (servers[i].ip == p->ip && servers[i].port == p->port)
            {
                LOG_INFO("%s delete from list ip %s port %d", __func__, str_ip(p->ip), p->port);
                p->port = 0;
                p->ip = 0;
                break;
            }
        }
    }

    // remove all items marked
    struct NewReplicaList head;
    head.next = first;

    struct NewReplicaList* pre = &head;
    for (p=first; p; )
    {
        struct NewReplicaList* next = p->next;
        if (p->ip == 0)
        {
            pre->next = next;
            free(p); // list node freed here, the node is in the new_replicas hash
        }
        else
        {
            pre = p;
        }
        p = next;
    }

    // if the 1st element is removed, update the hash by re-inserting
    if (head.next != first)
    {
        // if the first node freed above, cpu will run here
        // so here, just can't delete the list again
        // and we can't just remove all of the node in the list
        // so catch the point, and just do nothing
        // mod by zhangzm

        // &first to avoid destruction in foo_hashtable_remove then double free called,
        // &first means nothing
        foo_hashtable_remove(new_replicas, key, &first);
        if (head.next == NULL)	// if all items removed
            return false;
        first = head.next;
        foo_hashtable_put(new_replicas, key, first);
    }

    // add new replica(s) to request
    time_t hold_deadline = time(0) - NEW_RECORD_TIMEOUT;
    for (p = first; p; p = p->next)
    {
        LOG_INFO("%s new add ones chunk %08x.%d addr %s:%d",
                __func__, volumeid, index, str_ip(p->ip), p->port);
        if (unlikely(n == MAX_FOLLOWING_SERVERS))
        {
            LOG_ERROR("number of total replicas exceed MAX_FOLLOWING_SERVERS (%u)!",
                    MAX_FOLLOWING_SERVERS);
            break;
        }
        servers[n].port = p->port;
        servers[n].ip = p->ip;
        n++;

        if(p->breed_time < hold_deadline)
        {// breed time is old than dead line
            LOG_INFO("del timeout address chunk %08x.%d (%s:%d)",
                    volumeid, index, str_ip(p->ip), p->port);
            p->port = 0;
            p->ip = 0;
        }
    }

    return true;	// we have new replica(s)
}